package com.bigdata.sql

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

import java.util.Properties

/**
 * 总结：
 * SparkSQL支持非常丰富的数据源，可以从文件，db中读取数据
 * 写出到文件、db中
 * 数据分析：
 * 现在有一批文件，我想要从文件中进行计算并提取结果
 * demo：把excel中的数据导出为csv，然后交由spark来计算
 * Spark可以用来分析本地类似excel的表格文件,一个表格只能存储1G的数据
 *
 * 假如我现在有50G的数据，并且每天的增量数据也有50G
 * 现在想用要Spark对这些数据做分析
 * 每天的增量数据存储在HDFS中，使用Spark分析HDFS中的文本数据
 *
 * 我们的大数据一定是存储在HDFS中的，使用spark计算HDFS中的数据，
 * 直接基于HDFS开发，很繁琐，想要简化开发
 * 我想要直接给Spark一个SQL语句，那么Spark就知道去这个sql查询的表所对应的HDFS文件中查询
 * 后续的所有操作只需要写SQL即可。
 * 我们需要引入一个组件：hive 的 metastore 服务！！！！
 */
object Demo4DataSource {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("source")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /*
    大数据领域，历史数据一般会存储在HDFS，使用spark做数仓的离线分析
    数据源的来源就是HDFS，指定前缀hdfs://node01:9000即可
    这里就直接使用本地文件模拟
     */
    val studentDF: DataFrame = spark
      .read
      .format("csv")
      // csv文件只有数据，没有header，需要自己指定schema，指定列名和列的类型
      .schema("id STRING,name STRING,age INT,gender STRING, clazz STRING")
      .option("sep", ",")
      .load("data/students.txt")

    // studentDF.show()

    /**
     * 将数据保持为csv格式
     * 按照班级分组，得到班级的人数,并将结果写入到指定目录
     */
    val stuCountDF: DataFrame = studentDF
      .groupBy($"clazz")
      .agg(count("clazz") as "count")

    // 把结果进行保存，可以直接存储在HDFS
    stuCountDF
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
    // .save("data/csv_write")


    /**
     * json格式
     * spark 会自动将json中字段名和字段类型解析出来
     * json格式比csv格式占用的空间更大，在大数据场景下不适用json
     * json文件自带了描述，所以不需要再指定schema
     */
    val jsonDF: DataFrame = spark
      .read
      .format("json")
      .load("data/students.json")

    jsonDF.show(false)


    // 将数据保存为json格式，同样携带了字段的描述，
    // 优点：可读性更高  缺点：但占用的空间更大
    val jsonResultDF: DataFrame = jsonDF
      .groupBy($"clazz")
      .agg(count("clazz") as "count")
    // 写出为json格式
    jsonResultDF
      .write
      .format("json")
      .mode(SaveMode.Overwrite)
    // .save("data/json_write")

    /**
     * parquet: 带表结构的压缩格式，记事本打开时看不懂的
     * 压缩：时间换空间，
     * 时间成本和空间成本哪个大？
     * 空间 -> 磁盘的存储空间，白菜价
     * 时间 -> 缩减计算所消耗的时间，处理起来就要复杂的多
     * 更多的场景下我们追求的是时间而不是空间
     * 自带了schema
     */
    jsonResultDF
      .write
      .format("parquet")
      .mode(SaveMode.Overwrite)
    // .save("data/parquet_write")

    /**
     * 读取parquet
     * parquet格式的数据自带了表结构，不需要手动指定
     */
    val parDF: DataFrame = spark
      .read
      .format("parquet")
      .load("data/parquet_write")

    //    parDF.printSchema()
    //    parDF.show()

    /**
     * 读取JDBC中的数据, 经过分析写出到本地文件
     */
    val empDF: DataFrame = spark
      .read
      .format("jdbc")
      // url 指定到database
      .option("url", "jdbc:mysql://localhost:3306/news")
      // 指定到table
      .option("dbtable", "emp")
      .option("user", "root")
      .option("password", "root")
      .load()

//    empDF.printSchema()
//    empDF.show()

    // 对DataFrame中的数据做计算，并且保存到JDBC
    val result: Dataset[Row] = empDF
      .where($"deptno" === "20")

    // 把处理过的 Dataset 通过jdbc写出到表中
    val properties = new Properties()
    properties.setProperty("user", "root")
    properties.setProperty("password", "root")

    result
      .write
      .mode(SaveMode.Overwrite)
      .jdbc("jdbc:mysql://localhost:3306/news", "t_emp_result", properties);
  }
}
